package com.afdor.rws.event.driven.channel.local.consumer;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.core.extension.ExtensionLoader;
import com.afdor.rws.event.driven.channel.consumer.AbstractEventConsumer;
import com.afdor.rws.event.driven.channel.consumer.EventConsumer;
import com.afdor.rws.event.driven.channel.consumer.handler.EventRetryStrategy;
import com.afdor.rws.event.driven.channel.local.LocalObserverEventListener;
import com.afdor.rws.event.driven.channel.local.observer.ObserverEvent;
import com.afdor.rws.event.driven.channel.local.observer.ObserverEventSource;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Optional;

/**
 * 本地事件消费者
 *
 * @author 悭梵
 * @date Created in 2018-08-08 10:59
 */
@Slf4j
@NoArgsConstructor
@Extension(value = "local", order = Extension.NORMAL_ORDER - 100, scope = Extension.Scope.PROTOTYPE, interfaceClass = EventConsumer.class)
public class LocalEventConsumer extends AbstractEventConsumer<ObserverEvent> {

    @Getter
    private ObserverEventSource eventSource;

    public LocalEventConsumer(ObserverEventSource eventSource) {
        super();
        setEventSource(eventSource);
    }

    @Override
    public EventRetryStrategy getEventRetryStrategy() {
        EventRetryStrategy eventRetryStrategy = super.getEventRetryStrategy();
        if (eventRetryStrategy == null) {
            return ExtensionLoader.getInstance(EventRetryStrategy.class)
                    .newExtension("local", getEventSource());
        }
        return eventRetryStrategy;
    }


    @Override
    protected void doStart() {
        getEventSource().addEventListener(getEventSourceName(), getConsumerName(), new LocalObserverEventListener(this));
        super.doStart();
    }

    @Override
    protected void doStop() {
        getEventSource().clearEventListener();
        super.doStop();
    }

    public void setEventSource(ObserverEventSource eventSource) {
        this.eventSource = Optional.ofNullable(eventSource).orElseThrow(() -> new IllegalArgumentException("事件管理容器为空"));
    }

    @Override
    protected boolean hasEvent() {
        return getEventSource().hasEvent(getEventSourceName(), getConsumerName());
    }

    @Override
    protected ObserverEvent getEvent() {
        return getEventSource().getEvent(getEventSourceName(), getConsumerName());
    }

    @Override
    protected List<ObserverEvent> getEvents(int batchSize) {
        return getEventSource().getEvents(getEventSourceName(), getConsumerName(), batchSize);
    }

    @Override
    protected void doSuccess(ObserverEvent observerEvent, Object result) {
        observerEvent.complete(result);
    }

    @Override
    protected void doFailure(ObserverEvent observerEvent, String message) {
        observerEvent.failure(message);
    }

}
